package com.lzcu.maoyan.mapreduce.mapreduce.mapper;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.format.DateTimeParseException;
import java.io.IOException;
import java.time.temporal.ChronoField;


/**
 * @author WangLiang
 * @date `2024/3/25`
 */
public class MovieMapper extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] source = line.split("\001");
        if (source.length >= 10) {
            String keyName = source[2];
            String keyTime = source[1];
            if (isHourEven(keyTime)) {
                StringBuilder valueText = new StringBuilder();
                for (int i = 3; i <= 7; i++) {
                    if (i == 4){
                        String frequency = source[4];
                        if (Integer.parseInt(frequency) <= 10){
                            return;
                        }
                    }
                    String item = source[i];
                    if (item.contains("<")) {
                        item = "-0.01";
                    }
                    if (item.contains("--")) {
                        item = "-1";
                    }
                    valueText.append(item.replaceAll("%", "")).append("\001");
                }
                String releaseDays = source[8];
                String releaseDaysFormat = releaseDays.replaceAll("上映", "")
                        .replaceAll("天", "").replaceAll("点映", "-1")
                        .replaceAll("首日", "1").replaceAll("展映", "-2")
                        .replaceAll("零点场", "-3");
                valueText.append("".equals(releaseDaysFormat) ? "-4" : releaseDaysFormat);
                valueText.append("\001");
                String boxOffice = source[9];
                long boxOfficeDouble = 0;
                if (boxOffice.contains("万")) {
                    boxOfficeDouble = (long) (Double.parseDouble(boxOffice.replaceAll("万", "")) * 10000);
                } else if (boxOffice.contains("亿")) {
                    boxOfficeDouble = (long) (Double.parseDouble(boxOffice.replaceAll("亿", "")) * 100000000);
                }
                valueText.append(boxOfficeDouble);
                context.write(new Text(keyName + "\001" + keyTime), new Text(valueText.toString()));
            }
        }
    }

    public static boolean isHourEven(String dateTimeStr) {
        DateTimeFormatter formatter = new DateTimeFormatterBuilder()
                .appendPattern("yyyy-MM-dd HH:mm:ss")
                .optionalStart()
                .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
                .optionalEnd()
                .toFormatter();
        LocalDateTime dateTime = LocalDateTime.parse(dateTimeStr, formatter);

        int hour = dateTime.getHour();
        int minute = dateTime.getMinute();

        return hour % 6 == 0 && minute < 5;
    }

}
